package com.talent.service.iot.mqtt.handler;

import com.talent.service.iot.mqtt.config.Constant;
import com.talent.service.iot.mqtt.model.TlMqttSession;
import com.talent.service.iot.mqtt.model.request.TlMqttPubAckReq;
import com.talent.service.iot.mqtt.model.variable.TlMqttPubAckVariableHead;
import com.talent.service.iot.mqtt.service.IMqttMsgService;
import com.talent.service.iot.mqtt.service.IMqttSessionService;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import java.util.Set;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
public class TlMqttPubAckHandler extends SimpleChannelInboundHandler<TlMqttPubAckReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttPubAckHandler.class);


    private final IMqttSessionService sessionService;

    private final IMqttMsgService msgService;

    public TlMqttPubAckHandler(
                               IMqttSessionService sessionService,
                               IMqttMsgService msgService){
        this.sessionService = sessionService;
        this.msgService = msgService;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPubAckReq msg) throws Exception {

        logger.debug("in【PUBACK】 handler");
        Object clientId = ctx.channel().attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if(clientId==null){
            ctx.channel().close();
            return;
        }

        Mono<TlMqttSession> sessionMono = sessionService.find(clientId.toString());
        sessionMono.subscribe(session -> {
            if(session==null){
                ctx.channel().close();
                return;
            }
            TlMqttPubAckVariableHead variableHead = msg.getVariableHead();
            Integer messageId = variableHead.getMessageId();
            logger.debug("receive client 【{}】 messageId 【{}】 ack",clientId,messageId);
            //获取到会话 如果有会话 那么就将qos1的消息拿出来 这里确认后就移除

            //删除本地存储的消息
            msgService.clearPublishMessage(clientId.toString(),messageId);

            //todo 取消定时器里面定时发送的任务
            
            Set<Integer> publishMsgStore = session.getPublishMsgStore();
            //获取会话中保留的消息 然后删除
            publishMsgStore.remove(messageId);

        });

    }
}
